package com.cmnit.analysis.common

import java.sql.{Connection, PreparedStatement, SQLException}

import com.cmnit.analysis.util.{ConfigurationManager, DataSourceUtil, EnvUtil, ServiceUtil}
import org.apache.log4j.Logger
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.{DataFrame, Dataset, Row}

trait TDao {
  private val logger: Logger = Logger.getLogger(this.getClass)

  /**
   * spark执行sql语句
   *
   * @param sql 语句
   * @return
   */
  def sparkSQL(sql: String): DataFrame = {
    EnvUtil.take().sql(sql)
  }

  /**
   * 广播变量表
   *
   * @param sql 语句
   * @return
   */
  def broadcastSQL(sql: String): Broadcast[DataFrame] = {
    val broadFrame = EnvUtil.take().sql(sql)
    EnvUtil.take().sparkContext.broadcast(broadFrame)
  }

  /**
   * 注册UDF函数
   *
   * @param name 函数名
   */
  def registerUDF(name: String): Unit = {
    name match {
      case "getFeeFromType" => EnvUtil.take().udf.register("getFeeFromType", (mediaType: String, obuProvFeeSumAfter: String, feeSumLocalAfter: String) => ServiceUtil.getFeeFromType(mediaType, obuProvFeeSumAfter, feeSumLocalAfter))
      case "getFee" => EnvUtil.take().udf.register("getFee", (multiProvince: String, fee: String, splitProvince: String) => ServiceUtil.getFee(multiProvince, fee, splitProvince))
      case "getVehiclePlateMD5" => EnvUtil.take().udf.register("getVehiclePlateMD5", (vehiclePlate: String) => ServiceUtil.getVehiclePlateMD5(vehiclePlate))
    }
  }

  /**
   * 读取vertica数据并显示
   *
   * @param tableName 读取表名
   * @param acctName  时间字段
   * @param acctDate  查询日期
   * @return
   */
  def read(tableName: String, acctName: String, acctDate: String): Dataset[Row] = {
    EnvUtil.take().read
      .format("jdbc")
      .option("dbtable", tableName)
      .option("driver", ConfigurationManager.getProperty("vertica.driver"))
      .option("url", ConfigurationManager.getProperty("vertica.url"))
      .option("username", ConfigurationManager.getProperty("vertica.username"))
      .option("password", ConfigurationManager.getProperty("vertica.password"))
      .load
      .where(acctName + "='" + acctDate + "'")
  }

  /**
   * 删除vertica数据
   *
   * @param tableName 表名
   * @param acctName  时间字段
   * @param acctDate  账期
   * @return
   */
  def delete(tableName: String, acctName: String, acctDate: String): Unit = {
    val sql = "delete from " + tableName + " where " + acctName + " = '" + acctDate + "'"
    var pstmt: PreparedStatement = null
    var connect: Connection = null
    try {
      connect = DataSourceUtil.getConnection
      pstmt = connect.prepareStatement(sql)
      pstmt.execute
      logger.info("删除" + tableName + "表的账期：" + acctDate)
    } catch {
      case e: SQLException => e.printStackTrace()
    } finally {
      DataSourceUtil.closeResource(pstmt, connect)
    }
  }
}
